package com.own.business.sync.trigger.polling.limit.core;

import com.own.business.sync.trigger.base.bak.util.SyncTriggerBakUtil;
import com.own.business.sync.trigger.base.config.SyncTriggerPollingCondition;
import com.own.business.sync.trigger.base.entity.SyncTrigger;
import com.own.business.sync.trigger.base.method.ResolveMethodSpringHook;
import com.own.business.sync.trigger.base.property.SyncTriggerProperty;
import com.own.business.sync.trigger.base.service.SyncTriggerService;
import com.own.component.common.util.executor.ExecutorUtil;
import com.own.component.common.util.list.ListUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Conditional;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.HashSet;
import java.util.function.BiConsumer;
import java.util.function.Consumer;

/**
 * SyncTriggerPollingLimitCoreTask
 * <p>
 * 启动完成后开始执行
 *
 * @author chenxueli
 * @date 2024-01-08 15:48:43
 */
@Slf4j
@Component
@Conditional(SyncTriggerPollingCondition.class)
public class SyncTriggerPollingLimitCoreTask {

    private final Object lock = new Object();

    @Resource
    private SyncTriggerProperty syncTriggerProperty;
    @Resource
    private SyncTriggerService syncTriggerService;
    @Resource
    private SyncTriggerBakUtil syncTriggerBakUtil;
    @Resource
    private ResolveMethodSpringHook resolveMethodSpringHook;

    @Resource
    private ExecutorUtil executorUtil;

    @PostConstruct
    public void init() {
        if (syncTriggerProperty.getUseThread()) {
            log.info("使用多线程处理");
            if (syncTriggerProperty.getUseVirtualThread()) {
                log.info("使用虚拟线程处理");
                executorUtil.execute(() -> {
                    // 创建虚拟线程
                    var service = ExecutorUtil.newVirtualThreadPerTaskExecutor();
                    run((operation, data) -> service.submit(() -> {
                        try {
                            operation.accept(data);
                        } catch (Exception e) {
                            log.debug("处理异常", e);
                            syncTriggerBakUtil.save(data, e.getMessage());
                        }
                    }));
                });
            } else {
                run((operation, data) -> executorUtil.execute(() -> {
                    try {
                        operation.accept(data);
                    } catch (Exception e) {
                        log.debug("处理异常", e);
                        syncTriggerBakUtil.save(data, e.getMessage());
                    }
                }));
            }
        } else {
            log.info("不使用多线程处理");
            executorUtil.execute(() -> {
                run((operation, data) -> {
                    try {
                        operation.accept(data);
                    } catch (Exception e) {
                        log.debug("处理异常", e);
                        syncTriggerBakUtil.save(data, e.getMessage());
                    }
                });
            });
        }
    }

    /**
     * 轮询信息
     *
     * @param runnable 具体执行的方法
     */
    @SuppressWarnings("all")
    private void run(BiConsumer<Consumer<SyncTrigger>, SyncTrigger> runnable) {
        var max = syncTriggerProperty.getPollingTimeList().length - 1;
        var c = 0;
        while (true) {
            synchronized (lock) {
                // 查询当前备份库的数量信息
                var list = syncTriggerService.listForLimit();
                var size = list.size();
                if (size != 0) {
                    log.info("查询到{}条数据，开始处理", size);
                    var set = new HashSet<String>();
                    // 数据处理，如果同一条数据的修改出现过多次，只需要提交一次即可
                    list.forEach(data -> {
                        var key = data.getModuleName() + "_" + data.getType() + "_" + data.getTargetId() + "_" + data.getOldStatus() + "_" + data.getNewStatus();
                        if (set.contains(key)) {
                            return;
                        }
                        set.add(key);
                        var method = resolveMethodSpringHook.getMethod(data.getModuleName());
                        if (method != null) {
                            var operation = resolveMethodSpringHook.getMethod(data.getType(), method);
                            runnable.accept(operation, data);
                        } else {
                            syncTriggerBakUtil.save(data, "未找到处理方法");
                        }
                        log.debug("处理完成:{},{},{}", data.getModuleName(), data.getType(), data.getTargetId());
                    });
                    ListUtil.groupList(list, 100).forEach(values -> {
                        syncTriggerService.deleteByIds(values.stream().map(SyncTrigger::getId).toList());
                    });
                    log.info("完成处理{}条数据，结束处理", size);
                    c = 0;
                    // 如果处理的数据小于一半，开始休眠
                    if (size < syncTriggerProperty.getStoreLimit()) {
                        var timeout = syncTriggerProperty.getPollingTimeList()[0];
                        log.info("未查询到调整的数据，开始休眠" + timeout + "ms");
                        try {
                            lock.wait(timeout);
                        } catch (InterruptedException e) {
                            log.error("等待异常", e);
                        }
                    }
                } else {
                    var index = Math.min(c, max);
                    var timeout = syncTriggerProperty.getPollingTimeList()[index];
                    log.info("未查询到调整的数据，开始休眠" + timeout + "ms");
                    try {
                        lock.wait(timeout);
                        c++;
                    } catch (InterruptedException e) {
                        log.error("等待异常", e);
                    }
                }
            }
        }
    }

}
